import redis
import uuid
import time
import random
import json
from threading import Thread
from typing import Optional, List

# 连接 Redis（生产环境需配置连接池、密码、超时等）
redis_client = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True,
    socket_timeout=5
)

# 模拟数据库（实际为MySQL等，存储首页新闻）
mock_news_db = [
    {"id": 1, "title": "Redis分布式锁实战", "content": "详细讲解SETNX的使用...", "create_time": "2025-10-30"},
    {"id": 2, "title": "Python并发编程技巧", "content": "多线程与分布式锁结合...", "create_time": "2025-10-29"}
]


def acquire_lock(lock_key: str, expire_seconds: int = 10) -> Optional[str]:
    """
    基于SETNX获取分布式锁
    :param lock_key: 锁的key
    :param expire_seconds: 锁过期时间（防止死锁）
    :return: 锁的唯一标识（释放锁时校验），None表示获取失败
    """
    # 生成唯一标识（避免释放其他线程的锁）
    lock_value = str(uuid.uuid4())
    # SETNX + 过期时间（原子操作）：key不存在则设置，同时指定过期时间
    # redis的set命令中，nx=True表示SETNX，ex指定过期时间（秒）
    result = redis_client.set(
        name=lock_key,
        value=lock_value,
        nx=True,
        ex=expire_seconds
    )
    return lock_value if result else None


def release_lock(lock_key: str, lock_value: str) -> bool:
    """
    释放分布式锁（原子性校验并删除）
    :param lock_key: 锁的key
    :param lock_value: 加锁时的唯一标识
    :return: 释放成功返回True，否则False
    """
    # Lua脚本：先校验锁的value是否匹配，匹配则删除（保证原子性）
    lua_script = """
        if redis.call('get', KEYS[1]) == ARGV[1] then
            return redis.call('del', KEYS[1])
        else
            return 0
        end
    """
    # 执行脚本：KEYS[1]为lock_key，ARGV[1]为lock_value
    result = redis_client.eval(lua_script, 1, lock_key, lock_value)
    return result == 1


def get_news_from_db() -> List[dict]:
    """模拟从数据库查询首页新闻（耗时操作）"""
    print("[数据库] 开始查询首页新闻（模拟耗时）...")
    time.sleep(2)  # 模拟数据库查询耗时（如2秒）
    print("[数据库] 首页新闻查询完成")
    return mock_news_db


def get_home_news(retry_times: int = 3, retry_delay: float = 0.3) -> Optional[List[dict]]:
    """
    获取首页新闻（带缓存+分布式锁控制并发）
    :param retry_times: 获取锁失败时的重试次数
    :param retry_delay: 重试延迟基数（秒）
    :return: 新闻列表或None
    """
    cache_key = "cache:news:home"  # 缓存key：首页新闻
    lock_key = "lock:news:home"    # 锁key：与缓存key关联，控制首页新闻的加载

    # 1. 先查缓存，命中直接返回
    cached_news = redis_client.get(cache_key)
    if cached_news:
        print("[缓存命中] 直接返回首页新闻")
        return json.loads(cached_news)  # 反序列化（存储时用JSON）

    # 2. 缓存未命中，尝试获取锁（控制只有一个线程查库）
    lock_value = None
    for _ in range(retry_times):
        lock_value = acquire_lock(lock_key, expire_seconds=10)  # 锁过期时间>业务耗时（如2秒）
        if lock_value:
            print(f"[获取锁成功] 锁标识：{lock_value}，开始加载数据")
            break
        # 获取锁失败，随机延迟后重试（避免惊群效应）
        delay = retry_delay + random.uniform(0, 0.2)
        print(f"[获取锁失败] {delay:.2f}秒后重试")
        time.sleep(delay)
    else:
        # 重试次数耗尽，返回空（或降级处理，如返回默认数据）
        print("[获取锁失败] 重试次数耗尽，无法加载新闻")
        return None

    try:
        # 3. 二次检查缓存（防止等待锁期间其他线程已更新缓存）
        cached_news = redis_client.get(cache_key)
        if cached_news:
            print("[二次检查缓存命中] 返回已更新的缓存")
            return json.loads(cached_news)

        # 4. 锁持有者查询数据库
        news_list = get_news_from_db()
        if not news_list:
            print("[数据库] 未查询到首页新闻")
            return None

        # 5. 更新缓存（设置过期时间，避免缓存永久有效）
        # 缓存5分钟（300秒），实际根据业务设置
        redis_client.setex(
            name=cache_key,
            time=300,
            value=json.dumps(news_list)  # 用JSON序列化存储
        )
        print("[缓存更新] 首页新闻已写入缓存")
        return news_list

    finally:
        # 6. 释放锁（无论成功失败，必须释放，避免死锁）
        if lock_value:
            success = release_lock(lock_key, lock_value)
            print(f"[释放锁] {'成功' if success else '失败'}，锁标识：{lock_value}")


# 测试：多线程并发访问首页新闻
def test_concurrent_access():
    # 先清除缓存，模拟缓存失效场景
    redis_client.delete("cache:news:home")
    print("=== 缓存已清除，开始并发测试 ===")

    # 模拟10个并发请求
    threads = [Thread(target=get_home_news) for _ in range(10)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    print("=== 并发测试结束 ===")


if __name__ == "__main__":
    test_concurrent_access()